package com.muh.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CallbackConsumer {
    private static final String HOST_NAME = "39.104.74.101";
    private static final int PORT = 5672;
    private static final String USER_NAME = "admin";
    private static final String PASSWORD = "admin";
    private static final String VIRTUAL_HOST = "/mirror";//虚拟机

    private static final String EXCHANGE_NAME = "callbackExchange";
    private static final String ALTER_EXCHANGE_NAME = "alterExchange";
    private static final String QUEUE_NAME = "callbackQueue";

    public static void main(String[] args) throws Exception {
        //建立连接Factory
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST_NAME);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        factory.setVirtualHost(VIRTUAL_HOST);
        //建立连接
        Connection connection = factory.newConnection();
        //创建一个通道用来传输数据
        Channel channel = connection.createChannel();

        Map<String, Object> exchangeDeclareParams = new HashMap<String, Object>();
        exchangeDeclareParams.put("alternate-exchange", ALTER_EXCHANGE_NAME);
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, exchangeDeclareParams);
        channel.exchangeDeclare(ALTER_EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, exchangeDeclareParams);

        //申明一个对列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key1");

        //正常接收消息
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                Long deliveryTag = message.getEnvelope().getDeliveryTag();
                String correlationId = message.getProperties().getCorrelationId();
                System.out.println("received message consumerTag:" + consumerTag +
                        "; message:"+new String(message.getBody(), "UTF-8")+
                        "; deliveryTag="+deliveryTag+
                        "; correlationId="+correlationId);
                channel.basicAck(deliveryTag, false);
            }
        };
        //去控制台把对列删除掉触发这个callback
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("canceled message consumerTag:" + consumerTag);
            }
        };
        //关闭连接后会触发这个callback
        ConsumerShutdownSignalCallback consumerShutdownSignalCallback = new ConsumerShutdownSignalCallback() {
            @Override
            public void handleShutdownSignal(String consumerTag, ShutdownSignalException e) {
                System.out.println("consumer shutdown messagge consumerTag: " + consumerTag + ";Exception:" + e.getMessage());
            }
        };
        channel.basicConsume(QUEUE_NAME, deliverCallback, cancelCallback, consumerShutdownSignalCallback);
    }
}
